package com.lcifn.zk.console.support;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * curator zk connection impl
 */
public class CuratorZkConnection implements ZkConnection{

    private static final Logger logger = LoggerFactory.getLogger(CuratorZkConnection.class);

    /**
     * 检查zk连接是否可通
     *
     * @param zkAddress
     * @return
     */
    @Override
    public boolean checkConnection(String zkAddress) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(zkAddress, retryPolicy);
        client.start();
        try {
            // 连接阻塞
            client.blockUntilConnected(1, TimeUnit.SECONDS);
            // 获取连接状态
            ZooKeeper.States state = client.getZookeeperClient().getZooKeeper().getState();
            logger.info("zookeeper state :" + state);
            return state == ZooKeeper.States.CONNECTED;
        } catch (Exception e) {
            logger.error(String.format("连接zookeeper[%s]失败：%s", zkAddress, e.getMessage()));
            return false;
        } finally {
            client.close();
        }
    }

    /**
     * 创建zk连接
     *
     * @param zkAddress
     * @return
     */
    @Override
    public ZkSession connect(String zkAddress) {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        CuratorFramework client = CuratorFrameworkFactory.newClient(zkAddress, retryPolicy);
        client.start();
        return new CuratorZkSession(client);
    }

    /**
     * @return
     */
    @Override
    public boolean disconnect() {
        return false;
    }
}
